/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月16日 下午4:47:42
*/
package cn.com.guage.flink.sink;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import redis.clients.jedis.Jedis;

/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月16日 下午4:47:42
*/
public class GuageRedisSink extends RichSinkFunction<Tuple3<String, String, Long>> {

	/**
	 * 
	 */
	private static final long serialVersionUID = -8500985818571286012L;
	 private transient Jedis jedis;

	    @Override
	    public void open(Configuration config) {

//	        String host = parameters.getRequired("redis.host");
	        String host = "127.0.0.1";
//	        String password = parameters.get("redis.password", "");
	        Integer port = 6379;
	        Integer timeout = 5000;
	        jedis = new Jedis(host, port, timeout);
	    }


	    @Override
	    public void invoke(Tuple3<String, String, Long> value, Context context) throws Exception {
	        if (!jedis.isConnected()) {
	            jedis.connect();
	        }
	        //保存
	        jedis.hset(value.f0, value.f1, String.valueOf(value.f2));
	    }

	    @Override
	    public void close() throws Exception {
	        jedis.close();
	    }
}
